package org.walkerljl.commons.dts.defaults;

import org.walkerljl.commons.asyncparallel.CompletionServiceUtils;
import org.walkerljl.commons.asyncparallel.FutureUtils;
import org.walkerljl.commons.dts.DtsExecutor;
import org.walkerljl.commons.dts.DtsManager;
import org.walkerljl.commons.dts.DtsParam;
import org.walkerljl.commons.dts.DtsResult;
import org.walkerljl.commons.log.Logger;
import org.walkerljl.commons.log.LoggerFactory;
import org.walkerljl.commons.thread.ThreadPool;
import org.walkerljl.commons.util.ListUtils;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.CompletionService;
import java.util.concurrent.Future;

/**
 * DefaultDtsExecutor
 *
 * @author lijunlin
 */
public class DefaultDtsExecutor implements DtsExecutor {

    private static final Logger LOGGER = LoggerFactory.getLogger(DtsManager.class);
    private static final String MESSAGE_PREFIX = "DefaultDtsExecutor";

    /**
     * 多任务异步并行
     */
    private boolean multiTaskAsyncParallel = false;

    public DefaultDtsExecutor() {

    }

    public DefaultDtsExecutor(boolean multiTaskAsyncParallel) {
        this.multiTaskAsyncParallel = multiTaskAsyncParallel;
    }

    @Override
    public DtsResult prepare(List<DtsParam> params) {
        if (params == null || params.size() == 0) {
            return DtsResult.success();
        }
        try {
            for (DtsParam ele : params) {
                boolean prepareResult = ele.getHandler().prepare(ele.getParam());
                if (!prepareResult) {
                    return DtsResult.failure("Fail to prepare");
                }
            }
            return DtsResult.success();
        } catch(Exception e) {
            return DtsResult.failure(e.getMessage(), e);
        }
    }

    @Override
    public DtsResult commit(List<DtsParam> params) {
        if (params == null || params.isEmpty()) {
            return DtsResult.success();
        }
        if (params.size() == 1 || !multiTaskAsyncParallel) {
        	return sequenceExecute(params);
        } else {
            return parallelExecute(params);
        }
    }

    @Override
    public DtsResult rollback(List<DtsParam> params) {
        if (params == null || params.isEmpty()) {
            return DtsResult.success();
        }
        List<DtsParam> completedParams = new ArrayList<DtsParam>();
        try {
            for (DtsParam ele : params) {
                ele.getHandler().rollback(ele.getParam());
                completedParams.add(ele);
                traceRollback(ele);
            }
            return DtsResult.success();
        } catch(Exception e) {
            return DtsResult.failure(e.getMessage(), completedParams, e);
        }
    }

    /**
     * 顺序执行
     * @param params
     * @return
     */
    private DtsResult sequenceExecute(List<DtsParam> params) {
        List<DtsParam> completedParams = new ArrayList<DtsParam>();
        try {
            for (DtsParam ele : params) {
                Object result = ele.getHandler().commit(ele.getParam());
                ele.setResult(result);
                completedParams.add(ele);
                traceCommit(ele);
            }
            return DtsResult.success();
        } catch(Exception e) {
            return DtsResult.failure(e.getMessage(), completedParams, e);
        }
    }

    /**
     * 并行执行
     * @param params
     * @return
     */
    private DtsResult parallelExecute(final List<DtsParam> params) {
        CompletionService<Boolean> completionService = ThreadPool.wrapCompletionService();
        int taskAmount = 0;
        List<Future<?>> futures = ListUtils.newArrayList();
        final List<DtsParam> completedParams = new ArrayList<DtsParam>();
        try {
            for (final DtsParam ele : params) {
                futures.add(completionService.submit(new Callable<Boolean>() {
                    @Override
                    public Boolean call() throws Exception {
                        Object result = ele.getHandler().commit(ele.getParam());
                        ele.setResult(result);
                        traceCommit(ele);
                        completedParams.add(ele);
                        return true;
                    }
                }));
                taskAmount++;
            }
            CompletionServiceUtils.await(taskAmount, completionService);
            return DtsResult.success();
        } catch (Exception e) {
            return DtsResult.failure(e.getMessage(), completedParams, e);
        } finally {
            FutureUtils.cancel(futures, true);
        }
    }

    /**
     * 跟踪Commit
     * @param param
     */
    private void traceCommit(DtsParam param) {
        if (LOGGER.isDebugEnabled()) {
            LOGGER.debug(String.format("%s, threadId:%s-->Success to execute, param:%s",
                    new Object[]{MESSAGE_PREFIX, Thread.currentThread().getId(),
                    param.toString()}));
        }
    }

    /**
     * 跟踪Rollback
     * @param param
     */
    private void traceRollback(DtsParam param) {
        if (LOGGER.isDebugEnabled()) {
            LOGGER.debug(String.format("%s, threadId:{}-->Success to rollback, param:%s",
                    new Object[]{MESSAGE_PREFIX, Thread.currentThread().getId(),
            		param.toString()}));
        }
    }
}
